package com.bancomer.batch.impl;

import java.io.File;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.log4j.Logger;

import com.bancomer.batch.Constants;
import com.bancomer.batch.MapConstants;
import com.bancomer.batch.converters.AuditAlert;
import com.bancomer.batch.converters.FileToObjectsConverter;
import com.bancomer.batch.converters.FlatWormWriter;
import com.bancomer.batch.converters.ObjectsToAuditFormatConverter;
import com.bancomer.batch.converters.impl.AuditAlertImpl;
import com.bancomer.batch.pig.Pig;
import com.bancomer.batch.pojo.Alert;
import com.bancomer.batch.pojo.Detail;
import com.bancomer.batch.pojo.FileHelper;
import com.bancomer.batch.pojo.PigScript;
import com.bancomer.batch.pojo.Summary;
import com.bancomer.batch.pojo.SummaryDetail;
import com.bancomer.batch.pojo.ValidationResult;
import com.bancomer.batch.transferes.DownloadHdfs;
import com.bancomer.batch.transferes.UploadHdfs;
import com.bancomer.batch.util.FilesWriter;
import com.bancomer.batch.util.StringConverter;
import com.bancomer.batch.validators.GeneralValidator;
import com.bancomer.batch.validators.impl.ExistingFilesValidator;


public class BatchCotrollerImpl extends BatchControllerSupport  {
	
	@Override
	public void init() {
		try{
			Alert alert = (Alert)super.getBatchContext().getContextParam().get("alert");
			Logger logger = Logger.getLogger("MainBatchController");
			long time = new Date().getTime();
			logger.info("Iniciando el proceso de "+alert.getName());
			logger.info("ID del proceso "+time);
			super.getBatchContext().getContextParam().put("alertBeginning", time);
			String successFilePath = alert.getOutputFiles().get(0).getLocalPath();
			File successFile = new File(successFilePath+"SUCCESS"); 
			if(successFile.exists()==true){
				successFile.delete();
			}
		}
		catch(Exception e){
			super.getBatchContext().setMsgError(e.getMessage());
			super.getBatchContext().setContinueFlow(false);
			return;
		}
	}

	@Override
	public void validateFiles() {
		try{
			Alert alert = (Alert)super.getBatchContext().getContextParam().get("alert");
			// TODO Auto-generated method stub
			List<FileHelper> propertiesFiles = alert.getPropertiesFiles();
			List<FileHelper> hdfsFiles = alert.getHdfsFiles();
			ValidationResult result;
			Map<String,String> hdfsInputs = new HashMap<String,String>();
			for(FileHelper propertiesFile:propertiesFiles){
				GeneralValidator existingFileValidator = new ExistingFilesValidator();
				String fileName = propertiesFile.getName();
				propertiesFile.setName(StringConverter.convertPatternToDate(fileName));
				result = existingFileValidator.executeValidation(propertiesFile, null);
				if(result.getSucces()==false){
					super.getBatchContext().setMsgError(result.getMessage());
					super.getBatchContext().setContinueFlow(result.getSucces());
					return;
				}
				if(propertiesFile.getValidationClass()!=null && !propertiesFile.getValidationClass().isEmpty()){
					Class<GeneralValidator> userClass;
					try {
						for(String classValidator:propertiesFile.getValidationClass()){
							userClass = (Class<GeneralValidator>) Class.forName(classValidator);
							GeneralValidator validator= (GeneralValidator) userClass.newInstance();
							result = validator.executeValidation(propertiesFile, alert);
							super.getBatchContext().setContinueFlow(result.getSucces());
							if(result.getSucces()==false){
								super.getBatchContext().setMsgError(result.getMessage());
								super.getBatchContext().setContinueFlow(result.getSucces());
								return;
							}
						}
					} 
					catch (Exception e) {
						super.getBatchContext().setMsgError(e.getMessage());
						super.getBatchContext().setContinueFlow(false);
						return;
					}      
				}
			}
			int contadorLoads = 1;
			for(FileHelper fileHdfs:hdfsFiles){
				GeneralValidator existingFileValidator = new ExistingFilesValidator();
				String fileName = fileHdfs.getName();
				fileHdfs.setName(StringConverter.convertPatternToDate(fileName));
				result = existingFileValidator.executeValidation(fileHdfs, null);
				if(result.getSucces()==false){
					super.getBatchContext().setMsgError(result.getMessage());
					super.getBatchContext().setContinueFlow(result.getSucces());
					return;
				}
				String pathHadoop = fileHdfs.getHdfsPath().replaceAll("hdfs://.*?/","/");
				hdfsInputs.put("haddopFile"+(contadorLoads),pathHadoop+fileHdfs.getName());
				if(fileHdfs.getValidationClass()!=null && !fileHdfs.getValidationClass().isEmpty()){
					Class<GeneralValidator> userClass;
					try {
						for(String classValidator:fileHdfs.getValidationClass()){
							userClass = (Class<GeneralValidator>) Class.forName(classValidator);
							GeneralValidator validator= (GeneralValidator) userClass.newInstance();
							result = validator.executeValidation(fileHdfs, alert);
							super.getBatchContext().setContinueFlow(result.getSucces());
							if(result.getSucces()==false){
								super.getBatchContext().setMsgError(result.getMessage());
								super.getBatchContext().setContinueFlow(result.getSucces());
								return;
							}
						}
					} 
					catch (Exception e) {
						super.getBatchContext().setMsgError(e.getMessage());
						super.getBatchContext().setContinueFlow(false);
						return;
					}      
				}
				contadorLoads++;
			}
			Map<String,String> outputs = new HashMap<String,String>();
			PigScript script = alert.getPigScript();
			int i=0;
			for(FileHelper output:script.getOutputResultsFiles()){
				String fileName = output.getName();
				output.setName(StringConverter.convertPatternToDate(fileName));
				String pathHadoop = output.getHdfsPath().replaceAll("hdfs://.*?/","/");
				if(output.getProperties()!=null&&!output.getProperties().isEmpty()){
					
					if(output.getProperties().getProperty("outputType")!=null && !"".equals(output.getProperties().getProperty("outputType"))){
						
						outputs.put(output.getProperties().getProperty("outputType"),pathHadoop+output.getName());
					}
					else{
						outputs.put("output"+(i+1),pathHadoop+output.getName());
					}
				}
				outputs.put("output"+(i+1),pathHadoop+output.getName());
				i++;
			}
			FileHelper parametersFile = alert.getOutputFiles().get(0);
			FilesWriter.writePropertiesFile(parametersFile,hdfsInputs);
			FilesWriter.writePropertiesFile(parametersFile,outputs);
		}
		catch(Exception e){
			super.getBatchContext().setMsgError(e.getMessage());
			super.getBatchContext().setContinueFlow(false);
			return;
		}
	}

	@Override
	public void putFilesToHdfs() {
		// TODO Auto-generated method stub
		try{
			Alert alert = (Alert)super.getBatchContext().getContextParam().get("alert");
			List<FileHelper> hdfsFiles = alert.getHdfsFiles();
			for (FileHelper fileHdfs:hdfsFiles){
				UploadHdfs.uploadFilesHdfs(fileHdfs);
			}
		}
		catch(Exception e){
			super.getBatchContext().setMsgError(e.getMessage());
			super.getBatchContext().setContinueFlow(false);
			return;
		}		
	}

	@Override
	public void executePig() {
		// TODO Auto-generated method stub
		try{
			Alert alert = (Alert)super.getBatchContext().getContextParam().get("alert");
			Pig.executePigScript(alert);
		}
		catch(Exception e){
			super.getBatchContext().setMsgError(e.getMessage());
			super.getBatchContext().setContinueFlow(false);
			return;
		}	
	}

	@Override
	public void getFilesOnHdfs() {
		// TODO Auto-generated method stub
		try{
			Alert alert = (Alert)super.getBatchContext().getContextParam().get("alert");
			List<FileHelper> outputFiles = alert.getPigScript().getOutputResultsFiles();
			for (FileHelper output:outputFiles){
				DownloadHdfs.downloadFilesHdfs(output);
			}
		}
		catch(Exception e){
			super.getBatchContext().setMsgError(e.getMessage());
			super.getBatchContext().setContinueFlow(false);
			return;
		}	
	}



	@Override
	public void prepareResponseFiles() {
		// TODO Auto-generated method stub
		try{
			/*Alert alert = (Alert)super.getBatchContext().getContextParam().get("alert");
			FileToObjectsConverter converter = new FileToObjectsConverter();
			List<Summary> summaryObjects = converter.getObjectsFromSummary(alert);
			List<Detail> detailObjects = converter.getObjectsFromDetail(alert);
			ObjectsToAuditFormatConverter auditConverter = new ObjectsToAuditFormatConverter();
			auditConverter.convertToAuditSummary(summaryObjects, alert);
			auditConverter.convertToAuditDetail(detailObjects, alert);*/
			
			Alert alert = (Alert)super.getBatchContext().getContextParam().get("alert");
			MapConstants properties = MapConstants.getInstance(alert.getProperties().getProperty("constantsFile"));
			final Properties alertConstants = properties.getConstants();
			AuditAlert auditSumary = FileToObjectsConverter.getObjectsFromSummary(alert);
			AuditAlert auditDetail =  FileToObjectsConverter.getObjectsFromDetail(alert);
			SummaryDetail summaryDetail = new SummaryDetail();
			summaryDetail.setAuditAlertSummary(auditSumary);
			summaryDetail.setAuditAlertDetail(auditDetail);
			AuditAlert auditSummaryDetail = new AuditAlertImpl();
			auditSummaryDetail.addSummaryDetail(summaryDetail);
			FileHelper summaryFile=null;
			FileHelper detailFile=null;
			String outputPath = "";
			
			for(FileHelper file:alert.getOutputFiles()){
				if(file.getType().equals(alertConstants.getProperty("FILE_OUTPUT"))){
					if(file.getProperties().getProperty("outputType").equals(alertConstants.getProperty("OUTPUT_TYPE_SUMMARY"))){
						outputPath = file.getLocalPath();
						summaryFile = file;
					}
					if(file.getProperties().getProperty("outputType").equals(alertConstants.getProperty("OUTPUT_TYPE_DETAIL"))){
						detailFile=file;
						outputPath = file.getLocalPath();
					}
				}
			}
			summaryFile.setName(StringConverter.convertPatternToDate(summaryFile.getName()));
			detailFile.setName(StringConverter.convertPatternToDate(detailFile.getName()));
			
			FlatWormWriter.create(alert.getProperties().getProperty("constantsFile"),summaryFile.getProperties().getProperty("xmlOutputConfig"), summaryFile.getLocalPath()+summaryFile.getName(), auditSumary);
			FlatWormWriter.create(alert.getProperties().getProperty("constantsFile"),detailFile.getProperties().getProperty("xmlOutputConfig"), detailFile.getLocalPath()+detailFile.getName(), auditDetail);
	
			FilesWriter.writeFile("OK-"+Constants.alertDate,outputPath+"SUCCESS", false);
		}
		catch(Exception e){
			super.getBatchContext().setMsgError(e.getMessage());
			super.getBatchContext().setContinueFlow(false);
			return;
		}	
	}
	
	
	

}
